#include "mqttclient.h"
#include "transport.h"
#include "MQTTPacket.h"
#include "FreeRTOS.h"
#include "task.h"
#include "string.h"
#include "sockets.h"

#include "lwip/opt.h"

#include "lwip/sys.h"
#include "lwip/api.h"

#include "lwip/sockets.h"

#include "cJSON_Process.h"
//#include "bsp_dht11.h"

extern QueueHandle_t MQTT_Data_Queue;
MQTT_USER_MSG  mqtt_user_msg;
int32_t MQTT_Socket = 0;

void deliverMessage(MQTTString* TopicName, MQTTMessage* msg, MQTT_USER_MSG* mqtt_user_msg);


////////////////////////////////////////////////////////////////////////////////
/// @brief  MQTT_Connect
/// @note   Initialize Client and log in to the server.
/// @param  None.
/// @retval Network descriptor .>=0: Send successfully <0: Send failed
////////////////////////////////////////////////////////////////////////////////
uint8_t MQTT_Connect(void)
{
    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
    uint8_t buf[200];
    int buflen = sizeof(buf);
    int len = 0;
    data.clientID.cstring = CLIENT_ID;
    data.keepAliveInterval = KEEPLIVE_TIME;
    data.username.cstring = USER_NAME;
    data.password.cstring = PASSWORD;
    data.MQTTVersion = MQTT_VERSION;
    data.cleansession = 1;

    len = MQTTSerialize_connect((unsigned char*)buf, buflen, &data);

    transport_sendPacketBuffer(buf, len);


    if (MQTTPacket_read(buf, buflen, transport_getdata) == CONNACK) {
        unsigned char sessionPresent, connack_rc;
        if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) != 1 || connack_rc != 0) {
            printf("Unable to connect, the error code is: %d!\n", connack_rc);
            return Connect_NOK;
        }
        else {
            printf("The user name and secret key are successfully verified, and the MQTT connection is successful!\n");
            return Connect_OK;
        }
    }
    else
        printf("MQTT connection is not responding!\n");
    return Connect_NOTACK;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  MQTT_PingReq
/// @note   Send the MQTT heartbeat packet.
/// @param  None.
/// @retval >=0: Send successfully <0: Send failed.
////////////////////////////////////////////////////////////////////////////////
int32_t MQTT_PingReq(int32_t sock)
{
    int32_t len;
    uint8_t buf[200];
    int32_t buflen = sizeof(buf);
    fd_set readfd;
    struct timeval tv;
    tv.tv_sec = 5;
    tv.tv_usec = 0;

    FD_ZERO(&readfd);
    FD_SET(sock, &readfd);

    len = MQTTSerialize_pingreq(buf, buflen);
    transport_sendPacketBuffer(buf, len);

    if(select(sock + 1, &readfd, NULL, NULL, &tv) == 0)
        return -1;
    if(FD_ISSET(sock, &readfd) == 0)
        return -2;
    if(MQTTPacket_read(buf, buflen, transport_getdata) != PINGRESP)
        return -3;
    return 0;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  MQTTSubscribe
/// @note   subscription.
/// @param  None.
/// @param  int32_t sock:socket
/// @param  int8_t *topic:item
/// @param  enum QoS pos:type
/// @retval >=0: Send successfully <0: Send failed.
////////////////////////////////////////////////////////////////////////////////
int32_t MQTTSubscribe(int32_t sock, char* topic, enum QoS pos)
{
    static uint32_t PacketID = 0;
    uint16_t packetidbk = 0;
    int32_t conutbk = 0;
    uint8_t buf[100];
    int32_t buflen = sizeof(buf);
    MQTTString topicString = MQTTString_initializer;
    int32_t len;
    int32_t req_qos, qosbk;

    fd_set readfd;
    struct timeval tv;
    tv.tv_sec = 2;
    tv.tv_usec = 0;

    FD_ZERO(&readfd);
    FD_SET(sock, &readfd);

    topicString.cstring = (char*)topic;
    req_qos = pos;

    len = MQTTSerialize_subscribe(buf, buflen, 0, PacketID++, 1, &topicString, &req_qos);

    if(transport_sendPacketBuffer(buf, len) < 0)
        return -1;

    if(select(sock + 1, &readfd, NULL, NULL, &tv) == 0)
        return -2;

    if(FD_ISSET(sock, &readfd) == 0)
        return -3;

    if(MQTTPacket_read(buf, buflen, transport_getdata) != SUBACK)
        return -4;

    if(MQTTDeserialize_suback(&packetidbk, 1, &conutbk, &qosbk, buf, buflen) != 1)
        return -5;

    if((qosbk == 0x80) || (packetidbk != (PacketID - 1)))
        return -6;

    return 0;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  UserMsgCtl
/// @note   User handler function.
/// @param  MQTT_USER_MSG  *msg:Structure pointer.
/// @retval None.
////////////////////////////////////////////////////////////////////////////////
void UserMsgCtl(MQTT_USER_MSG*  msg)
{

    printf("*****Subscribed! ******\n");
    if(msg->msglenth > 2) {
        switch(msg->msgqos) {
            case 0:
                printf("MQTT>>Quality:QoS0\n");
                break;
            case 1:
                printf("MQTT>>Quality:QoS1\n");
                break;
            case 2:
                printf("MQTT>>Quality:QoS2\n");
                break;
            default:
                printf("MQTT>>Wrong quality\n");
                break;
        }
        printf("MQTT>>topic:%s\n", msg->topic);
        printf("MQTT>>Content:%s\n", msg->msg);
        printf("MQTT>>length:%d\n", msg->msglenth);
        Proscess(msg->msg);
    }
    msg->valid  = 0;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  GetNextPackID
/// @note   Generates the next packet ID.
/// @param  None.
/// @retval None.
////////////////////////////////////////////////////////////////////////////////
uint16_t GetNextPackID(void)
{
    static uint16_t pubpacketid = 0;
    return pubpacketid++;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  mqtt_msg_publish
/// @note   User push.
/// @param  None.
/// @retval None.
////////////////////////////////////////////////////////////////////////////////
int32_t MQTTMsgPublish(int32_t sock, char* topic, int8_t qos, uint8_t* msg)
{
    int8_t retained = 0;
    uint32_t msg_len = 0;
    uint8_t buf[MSG_MAX_LEN];
    int32_t buflen = sizeof(buf), len;
    MQTTString topicString = MQTTString_initializer;
    uint16_t packid = 0, packetidbk = 0;

    topicString.cstring = (char*)topic;

    if((qos == QOS1) || (qos == QOS2)) {
        packid = GetNextPackID();
    }
    else {
        qos = QOS0;
        retained = 0;
        packid = 0;
    }
    msg_len = strlen((char*)msg_len);

    len = MQTTSerialize_publish(buf, buflen, 0, qos, retained, packid, topicString, (unsigned char*)msg, msg_len);
    if(len <= 0)
        return -1;
    if(transport_sendPacketBuffer(buf, len) < 0)
        return -2;

    if(qos == QOS0) {
        return 0;
    }
    if(qos == QOS1) {
        if(WaitForPacket(sock, PUBACK, 5) < 0)
            return -3;
        return 1;
    }
    if(qos == QOS2) {

        if(WaitForPacket(sock, PUBREC, 5) < 0)
            return -3;

        len = MQTTSerialize_pubrel(buf, buflen, 0, packetidbk);
        if(len == 0)
            return -4;
        if(transport_sendPacketBuffer(buf, len) < 0)
            return -6;

        if(WaitForPacket(sock, PUBREC, 5) < 0)
            return -7;
        return 2;
    }
    return -8;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  ReadPacketTimeout
/// @note   Blocks reading MQTT data.
/// @param  None.
/// @param  int32_t sock:Network descriptor
/// @param  uint8_t *buf:buff
/// @param  int32_t buflen:length
/// @param  uint32_t timeout: timeout
/// @retval Package type.
////////////////////////////////////////////////////////////////////////////////
int32_t ReadPacketTimeout(int32_t sock, uint8_t* buf, int32_t buflen, uint32_t timeout)
{
    fd_set readfd;
    struct timeval tv;
    if(timeout != 0) {
        tv.tv_sec = timeout;
        tv.tv_usec = 0;
        FD_ZERO(&readfd);
        FD_SET(sock, &readfd);


        if(select(sock + 1, &readfd, NULL, NULL, &tv) == 0)
            return -1;

        if(FD_ISSET(sock, &readfd) == 0)
            return -1;
    }

    return MQTTPacket_read(buf, buflen, transport_getdata);
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  initialize SPI2 MODE1(master)
/// @note   Receive data from the server.
/// @param  MQTTMessage *msg:MQTT
/// @param  MQTT_USER_MSG *mqtt_user_msg:The user accepts the structure
/// @param  MQTTString  *TopicName:theme.
/// @retval None.
////////////////////////////////////////////////////////////////////////////////
void deliverMessage(MQTTString*  TopicName, MQTTMessage* msg, MQTT_USER_MSG* mqtt_user_msg)
{

    mqtt_user_msg->msgqos = msg->qos;

    memcpy(mqtt_user_msg->msg, msg->payload, msg->payloadlen);
    mqtt_user_msg->msg[msg->payloadlen] = 0;

    mqtt_user_msg->msglenth = msg->payloadlen;

    memcpy((char*)mqtt_user_msg->topic, TopicName->lenstring.data, TopicName->lenstring.len);
    mqtt_user_msg->topic[TopicName->lenstring.len] = 0;

    mqtt_user_msg->packetid = msg->id;

    mqtt_user_msg->valid = 1;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  mqtt_pktype_ctl
/// @note   Processes according to the package type.
/// @param  packtype :package type.
/// @param  buf      :buff.
/// @param  buflen   :length.
/// @retval None.
////////////////////////////////////////////////////////////////////////////////
void mqtt_pktype_ctl(uint8_t packtype, uint8_t* buf, uint32_t buflen)
{
    MQTTMessage msg;
    int32_t rc;
    MQTTString receivedTopic;
    uint32_t len;
    switch(packtype) {
        case PUBLISH:

            if(MQTTDeserialize_publish(&msg.dup, (int*)&msg.qos, &msg.retained, &msg.id, &receivedTopic,
                                       (unsigned char**)&msg.payload, &msg.payloadlen, buf, buflen) != 1)
                return;

            deliverMessage(&receivedTopic, &msg, &mqtt_user_msg);


            if(msg.qos == QOS0) {


                UserMsgCtl(&mqtt_user_msg);
                return;
            }

            if(msg.qos == QOS1) {
                len = MQTTSerialize_puback(buf, buflen, mqtt_user_msg.packetid);
                if(len == 0)
                    return;

                if(transport_sendPacketBuffer(buf, len) < 0)
                    return;

                UserMsgCtl(&mqtt_user_msg);
                return;
            }


            if(msg.qos == QOS2) {
                len = MQTTSerialize_ack(buf, buflen, PUBREC, 0, mqtt_user_msg.packetid);
                if(len == 0)
                    return;

                transport_sendPacketBuffer(buf, len);
            }
            break;
        case  PUBREL:

            rc = MQTTDeserialize_ack(&msg.type, &msg.dup, &msg.id, buf, buflen);
            if((rc != 1) || (msg.type != PUBREL) || (msg.id != mqtt_user_msg.packetid))
                return ;

            if(mqtt_user_msg.valid == 1) {

                UserMsgCtl(&mqtt_user_msg);
            }

            len = MQTTSerialize_pubcomp(buf, buflen, msg.id);
            if(len == 0)
                return;

            transport_sendPacketBuffer(buf, len);
            break;
        case   PUBACK:
            break;
        case   PUBREC:
            break;
        case   PUBCOMP:
            break;
        default:
            break;
    }
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  WaitForPacket
/// @note   None.
/// @param  sock        Network descriptor.
/// @param  packettype  Package type.
/// @param  times       times.
/// @retval >=0: wait for specific package <0: did not wait for specific package.
////////////////////////////////////////////////////////////////////////////////
int32_t WaitForPacket(int32_t sock, uint8_t packettype, uint8_t times)
{
    int32_t type;
    uint8_t buf[MSG_MAX_LEN];
    uint8_t n = 0;
    int32_t buflen = sizeof(buf);
    do {

        type = ReadPacketTimeout(sock, buf, buflen, 2);
        if(type != -1)
            mqtt_pktype_ctl(type, buf, buflen);
        n++;
    } while((type != packettype) && (n < times));

    if(type == packettype)
        return 0;
    else
        return -1;
}



void Client_Connect(void)
{
    char* host_ip;

#if  LWIP_DNS
    ip4_addr_t dns_ip;
    netconn_gethostbyname(HOST_NAME, &dns_ip);
    host_ip = ip_ntoa(&dns_ip);
    PRINT_DEBUG("host name : %s , host_ip : %s\n", HOST_NAME, host_ip);
#else
    host_ip = HOST_NAME;
#endif
MQTT_START:


    printf("1.Start to connect to the server of the corresponding cloud platform...\n");
    printf("Server IP address: %s, port number:%0d!\n", host_ip, HOST_PORT);
    while(1) {

        MQTT_Socket = transport_open((int8_t*)host_ip, HOST_PORT);

        if(MQTT_Socket >= 0) {
            printf("Successfully connected to the cloud platform server!\n");
            break;
        }
        printf("Failed to connect to the cloud platform server, wait 3 seconds and try to connect again!\n");

        vTaskDelay(3000);
    }

    printf("2. MQTT user name and key authentication login...\n");

    if(MQTT_Connect() != Connect_OK) {

        printf("MQTT user name and key verification login failed...\n");

        transport_close();
        goto MQTT_START;
    }
    printf("3. Start subscription...\n");
    if(MQTTSubscribe(MQTT_Socket, (char*)TOPIC, QOS1) < 0) {

        printf("Client subscription failed...\n");

        transport_close();
        goto MQTT_START;
    }
    printf("4. Start cyclically receiving subscriptions...\n");
}

////////////////////////////////////////////////////////////////////////////////
/// @brief  mqtt_thread
/// @note   MQTT cloud connection steps:
///         1. Connect to the server of the corresponding cloud platform
///         2.MQTT user authenticates login with secret key
///         3. Subscribe to the specified topic
///         4. Waiting for the data of the received topic and the data of the submitted topic.
/// @param  The task parameters.
/// @retval None.
////////////////////////////////////////////////////////////////////////////////
void mqtt_thread(void* pvParameters)
{
    uint32_t curtick;
    uint8_t no_mqtt_msg_exchange = 1;
    uint8_t buf[MSG_MAX_LEN];
    int32_t buflen = sizeof(buf);
    int32_t type;
    fd_set readfd;
    struct timeval tv;
    tv.tv_sec = 0;
    tv.tv_usec = 10;

MQTT_START:

    Client_Connect();
    curtick = xTaskGetTickCount();
    while(1) {
        no_mqtt_msg_exchange = 1;

        FD_ZERO(&readfd);
        FD_SET(MQTT_Socket, &readfd);

        select(MQTT_Socket + 1, &readfd, NULL, NULL, &tv);

        if(FD_ISSET(MQTT_Socket, &readfd) != 0) {

            type = ReadPacketTimeout(MQTT_Socket, buf, buflen, 50);
            if(type != -1) {
                mqtt_pktype_ctl(type, buf, buflen);
                no_mqtt_msg_exchange = 0;
                curtick = xTaskGetTickCount();
            }
        }

        if((xTaskGetTickCount() - curtick) > (KEEPLIVE_TIME / 2 * 1000)) {
            curtick = xTaskGetTickCount();
            if(no_mqtt_msg_exchange == 0) {

                continue;
            }
            if(MQTT_PingReq(MQTT_Socket) < 0) {
                printf("Send keep-alive ping failed....\n");
                goto CLOSE;
            }
            printf("Send keep-alive ping as heartbeat success....\n");
            no_mqtt_msg_exchange = 0;
        }
    }

CLOSE:
    transport_close();
    goto MQTT_START;
}

void mqtt_send(void* pvParameters)
{
    int32_t ret;
    uint8_t no_mqtt_msg_exchange = 1;
    uint32_t curtick;
    uint8_t res;

    BaseType_t xReturn = pdTRUE;
//    uint32_t* r_data;
    DHT11_Data_TypeDef* recv_data;

    cJSON* cJSON_Data = NULL;
    cJSON_Data = cJSON_Data_Init();
    double a, b;
MQTT_SEND_START:

    while(1) {
        xReturn = xQueueReceive( MQTT_Data_Queue,
                                 &recv_data,
                                 3000);
        if(xReturn == pdTRUE) {
            a = recv_data->temperature;
            b = recv_data->humidity;
//        printf("a = %f,b = %f\n",a,b);

            res = cJSON_Update(cJSON_Data, TEMP_NUM, &a);
            res = cJSON_Update(cJSON_Data, HUM_NUM, &b);

            if(UPDATE_SUCCESS == res) {

                char* p = cJSON_Print(cJSON_Data);

                ret = MQTTMsgPublish(MQTT_Socket, (char*)TOPIC, QOS0, (uint8_t*)p);
                if(ret >= 0) {

                    no_mqtt_msg_exchange = 0;

                    curtick = xTaskGetTickCount();
                }
                vPortFree(p);
                p = NULL;
            }
            else
                printf("update fail\n");
        }

        if((xTaskGetTickCount() - curtick) > (KEEPLIVE_TIME / 2 * 1000)) {
            curtick = xTaskGetTickCount();

            if(no_mqtt_msg_exchange == 0) {

                continue;
            }

            if(MQTT_PingReq(MQTT_Socket) < 0) {

                printf("Send keep-alive ping failed....\n");
                goto MQTT_SEND_CLOSE;
            }
            printf("Send keep-alive ping as heartbeat success....\n");
            no_mqtt_msg_exchange = 0;
        }
    }
MQTT_SEND_CLOSE:

    transport_close();

    Client_Connect();
    goto MQTT_SEND_START;
}

void
mqtt_thread_init(void)
{
    sys_thread_new("iperf_client", mqtt_thread, NULL, 1024, 6);
    sys_thread_new("iperf_client", mqtt_send, NULL, 1024, 7);
}

